/// Sink[T] accepts values of type T.
type Sink[in T]
    native; // rx.Sink

/// Source[T] represents an effect that subscribes to a discrete data source.
/// It is a subtype of Observable[T]. Different from general Observable, a Source
/// never terminates.
type Source[out T]
    protected
    Observable[T];

/// Bus[T] accepts values of type T and provides accepted T values,
/// similar to the concept of Subject in ReactiveX.
type Bus[T]
    native; // rx.Bus
    method Bus.in:  Sink[T]    native '__BusIn';
    method Bus.out: Source[T]  native '__BusOut';

/// Reactive[T] represents a reactively mutable T value,
/// similar to the concept of BehaviourSubject in ReactiveX.
type Reactive[T]
    protected Bus[T]; // rx.Reactive
    method Reactive.in:  Sink[T]      native '__BusIn';
    method Reactive.out: Computed[T]  native '__BusOut';


export const Blackhole: Sink[any]
    native '__Blackhole';

export function Sink:[T]
    &(&(T) => Async) => Sink[T]
    native '__Sink';

export function Source:[T]
    &(Observable[T]) => Source[T]
    native '__SourceFromObservable';

export function Bus:[T]
    &() => Sync[Bus[T]]
    native '__Bus';

export function Reactive:[T]
    & { initial: T } => Sync[Reactive[T]]
    native '__Reactive';

export function Reactive:[T]
    & { in: Sink[T], out: Computed[T] } => Reactive[T]
    native '__ReactiveFromSinkComputed';


export function <- :
    [T] &(Sink[T], T) => Sync
    native '__SinkEmit';

export function adapt:[A,B]
    &(Sink[A], &(B) => A) => Sink[B]
    native '__SinkAdapt';

export function debounce:[T]
    &(Sink[T], Number) => Sync[Sink[T]]
    native '__SinkDebounce';


export function ->:[T]
    &(Source[T], Sink[T]) => Async[never]
    native '__Connect';

export function ->:[T]
    &(Source[T], &(T) => Sync) => Async[never]
    &(source, receiver) => (source -> {Sink(receiver)});

export function ->:
    &(Source[unit], Sync) => Async[never]
    &(source, receiver) => (source -> {Sink::[unit](&() => receiver)});

export function map:[A,B]
    &(Source[A], &(A) => B) => Source[B]
    native '__ObservableMap';

export function filter:[T]
    &(Source[T], &(T) => Bool) => Source[T]
    native '__ObservableFilter';

export function scan:[T,A]
    &(Source[T], (A, &(A,T) => A)) => Source[A]
    native '__ObservableScan';

export function debounce:[T]
    &(Source[T], Number) => Source[T]
    native '__ObservableDebounce';

export function merge:[T]
    &(List[Source[T]]) => Source[T]
    native '__ObservableMerge';

export function concat-map:[A,B]
    &(Source[A], &(A) => Observable[B]) => Source[B]
    native '__ObservableConcatMap';

export function switch-map:[A,B]
    &(Source[A], &(A) => Observable[B]) => Source[B]
    native '__ObservableSwitchMap';

export function exhaust-map:[A,B]
    &(Source[A], &(A) => Observable[B]) => Source[B]
    native '__ObservableExhaustMap';

function distinct-until-changed:[T]
    &(Source[T], &(T,T) => Bool) => Source[T]
    native '__ObservableDistinctUntilChanged';

export function distinct-until-changed:
    [T] { = : Op=[T] }
    &(Source[T]) => Source[T]
    &(a) => { distinct-until-changed (a, =) };


export function update:[T]
    &(Reactive[T], &(T) => T) => Sync
    native '__ReactiveUpdate';

function morph:[A,B]
    &(Reactive[A], (&(A) => &(B) => A), (&(A) => B)) => Reactive[B]
    native '__ReactiveMorph';

export function morph:[A,B]
    &(Reactive[A], { in: &(A) => &(B) => A, out: &(A) => B }) => Reactive[B]
    &(r, f) => { morph (r, f.in, f.out) };

export function proj:[A,B]
    &(Reactive[A], &(A) => ProjRef[A,B]) => Reactive[B]
    &(r, ref) => (r morph {
        in:  (&a => &b => {{ref a}.write b}),
        out: (&a => {ref a}.read)
    });

export function proj?:[A,B]
    &(Reactive[A], &(A) => CaseRef[A,B]) => Reactive[Maybe[B]]
    &(r, ref) => (r morph {
        in:  (&a => &b? => let (a,_) := { {ref a} b? }, a),
        out: (&a => a.{ref}.read)
    });

export function proj-list:[A,B]
    &(Reactive[A], &(A) => ProjRef[A,List[B]]) => Sync[ListEdit[B]]
    &(r, ref) => { ListEdit (r proj ref) };


